package com.example.demo.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * <p> 消息消费者步骤分析
 * 1.创建消费者Consumer，制定消费者组名
 * 2.指定Nameserver地址
 * 3.订阅主题Topic和Tag
 * 4.设置回调函数，处理消息
 * 5.启动消费者consumer
 * </p>
 *
 * @author Jon
 * @since 2021/7/13 16:35
 */
public class Consumer {
    
    /*
     * @description 消费者采用负载均衡方式消费消息，多个消费者共同消费队列消息，每个消费者处理的消息不同
     * @param       [args]
     * @return      void
     * @author      Jon
     * @since       2021/7/13 14:12
     */
//    public static void main(String[] args) throws  Exception {
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroup");
//        consumer.setNamesrvAddr("127.0.0.1:9876");
//        //消费者订阅topic
//        consumer.subscribe("topicName","*");
//        //负载均衡模式消费
//        consumer.setMessageModel(MessageModel.CLUSTERING);
//        // 注册回调函数，处理消息
//        consumer.registerMessageListener(new MessageListenerConcurrently(){
//            @Override
//            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
//                System.out.println(list);
//                System.out.println(consumeConcurrentlyContext);
//                return null;
//            }
//        });
//        consumer.start();
//        System.out.println("消费者启动");
//    }

    /*
     * @description 消费者采用广播的方式消费消息，每个消费者消费的消息都是相同的
     * @param       [args]
     * @return      void
     * @author      Jon
     * @since       2021/7/13 17:49
     */
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.subscribe("topic_cib_hr","*");
        //广播模式消费
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("consumer start");
    }

 }
